package rpc.client.netty;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;
import rpc.common.enumeraion.PackageType;
import rpc.common.utils.SingletonFactory;
import rpc.packet.RpcMessage;
import rpc.packet.RpcResponse;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    public static final Map<String, Promise<Object>> PROMISE_MAP = new ConcurrentHashMap<>();

    private final UnprocessedRequests unprocessedRequests;

    public NettyClientHandler() {
        unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            log.debug("msg: {}", msg);
            if (msg instanceof RpcMessage) {
                RpcMessage rpcMessage = (RpcMessage) msg;
                Byte packageType = rpcMessage.getPackageType();
                if (packageType == PackageType.RESPONSE_PACK.getCode()) {
                    RpcResponse response = (RpcResponse) rpcMessage.getData();
                    //返回结果
                    unprocessedRequests.complete(response);
                }
            }
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //连接建立后触发active
        log.debug("服务器连接已建立");
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.debug("连接已经断开");
        super.channelInactive(ctx);
    }
}
